package requestCase

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Req01_re318 {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val context = new SparkContext(conf)

    val logs = context.textFile("D:\\IDEA_prjects\\sparkDemo\\src\\main\\scala\\data\\user_visit_action.txt")
    logs.cache()

    /**
     * click
     */
    val clickRDD = logs.filter(
      line => {
        val data = line.split("_")
        data(6) != "-1"
      }
    )
    val clickRDDcount = clickRDD.map(
      data => {
        val strings = data.split("_")
        (strings(6), 1)
      }
    ).reduceByKey(_ + _)

    /**
     * order
     */
    val odrerRDD = logs.filter(
      line => {
        val strings = line.split("_")
        strings(8) != "null"
      }
    )
    val orderRDDcount = odrerRDD.flatMap(
      data => {
        val strings = data.split("_")
        val strings1 = strings(8).split(",")
        strings1.map(
          (_, 1)
        )
      }
    ).reduceByKey(_ + _)

    /**
     * pay
     */
    val payRDD = logs.filter(
      line => {
        val strings = line.split("_")
        strings(10) != "null"
      }
    )
    val payRDDcount = payRDD.flatMap(
      data => {
        val strings = data.split("_")
        val strings1 = strings(10).split(",")
        strings1.map((_, 1))
      }
    ).reduceByKey(_ + _)

    /**
     * 聚合结果
     */

    val cogroupRDD = clickRDDcount.cogroup(orderRDDcount, payRDDcount)

    val result: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues {
      case (click, order, pay) => {
        var clickCnt = 0
        val click_iterator = click.iterator
        if (click_iterator.hasNext) {
          clickCnt = click_iterator.next()
        }
        var orderCnt = 0
        val order_iterator = order.iterator
        if (order_iterator.hasNext) {
          orderCnt = order_iterator.next()
        }
        var payCnt = 0
        val pay_iterator = pay.iterator
        if (pay_iterator.hasNext) {
          payCnt = pay_iterator.next()
        }
        (clickCnt, orderCnt, payCnt)
      }
    }

    val result_tuples: Array[(String, (Int, Int, Int))] = result.sortBy(_._2, false).take(10)

    result_tuples.foreach(println)

    context.stop()

  }

}
